package etcddriver

import (
	"context"
	"encoding/json"
	"fmt"
	"path"
	"sort"
	"strings"
	"sync"
	"time"

	etcdClient "go.etcd.io/etcd/client/v3"
	// etcdClient "github.com/coreos/etcd/clientv3"
	hash "github.com/mitchellh/hashstructure"
	"github.com/sirupsen/logrus"
)

var (
	DefaultAddr    = "127.0.0.1:2379"
	DefaultTimeout = 5 * time.Second

	DefaultPrefix = "/agent/register"
)

type EtcdDriver struct {
	ctx         context.Context
	etcdClient  *etcdClient.Client
	Endpoints   []string
	DialTimeout int
	ReqTimeout  int
	prefix      string
	LeaseID     etcdClient.LeaseID
	Lease       etcdClient.Lease
	sync.RWMutex
	register map[string]uint64
	leases   map[string]etcdClient.LeaseID
}

type Service struct {
	Name    string  `json:"name"`
	Version string  `json:"version"`
	Nodes   []*Node `json:"nodes"`
}

type Node struct {
	Id       string            `json:"id"`
	Address  string            `json:"address"`
	Metadata map[string]string `json:"metadata"`
}

func NewEtcdDriver(ctx context.Context, dialTimeout int, reqTimeout int, endPoints []string) (*EtcdDriver, error) {
	cli, err := etcdClient.New(etcdClient.Config{
		Endpoints:   endPoints,
		DialTimeout: time.Duration(dialTimeout) * time.Second,
	})

	if err != nil {
		return nil, fmt.Errorf("create etcd client failed, err: %v", err)
	}

	fmt.Println("create etcd client success.")

	return &EtcdDriver{
		ctx:         ctx,
		Endpoints:   endPoints,
		etcdClient:  cli,
		DialTimeout: dialTimeout,
		ReqTimeout:  reqTimeout,
		prefix:      DefaultPrefix,
		register:    make(map[string]uint64),
		leases:      make(map[string]etcdClient.LeaseID),
	}, nil
}

func (t *EtcdDriver) Close() error {
	return t.etcdClient.Close()
}

func (t *EtcdDriver) Get(key string) (bool, []byte) {
	ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
	defer cancel()
	resp, err := t.etcdClient.Get(ctx, key)
	if err != nil {
		fmt.Printf("get from etcd failed, err:%v\n", err)
		return false, nil
	}

	if len(resp.Kvs) > 1 {
		fmt.Printf("get key: %s failed, err: have multi values, maybe can use prefix", key)
		return false, nil
	}
	if len(resp.Kvs) == 0 {
		fmt.Printf("get key: %s failed, err: have no value, check your key string", key)
		return false, nil
	}
	kv := resp.Kvs[0]
	return true, kv.Value
}

func (t *EtcdDriver) GetPrefix(key string) (bool, map[string][]byte) {
	resultMap := make(map[string][]byte)
	ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
	defer cancel()
	resp, err := t.etcdClient.Get(ctx, key, etcdClient.WithPrefix())
	if err != nil {
		fmt.Printf("get key: %s failed, err: %v", key, err)
		return false, nil
	}
	if len(resp.Kvs) == 0 {
		fmt.Printf("get key: %s failed, err: have no value, check your key string", key)
		return false, nil
	}
	for _, kv := range resp.Kvs {
		resultMap[string(kv.Key)] = kv.Value
	}

	return true, resultMap
}

func (t *EtcdDriver) Put(key, value string) error {
	ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
	defer cancel()
	_, err := t.etcdClient.Put(ctx, key, value)
	if err != nil {
		return fmt.Errorf("put key: %s failed,err: %v", key, err)
	}
	return nil
}

func (t *EtcdDriver) Del(key string) error {
	ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
	defer cancel()
	_, err := t.etcdClient.Delete(ctx, key)
	if err != nil {
		return fmt.Errorf("del key: %s failed,err: %v", key, err)
	}
	return nil
}

func (t *EtcdDriver) DelPrefix(key string) error {
	ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
	defer cancel()
	_, err := t.etcdClient.Delete(ctx, key, etcdClient.WithPrefix())
	if err != nil {
		return fmt.Errorf("del key prefix: %s failed,err: %v", key, err)
	}
	return nil
}

func (t *EtcdDriver) WatchKey(key string) etcdClient.WatchChan {
	ch := t.etcdClient.Watch(t.ctx, key)
	for wresp := range ch {
		for _, evt := range wresp.Events {
			fmt.Printf("Type:%v key:%v value:%v\n", evt.Type, string(evt.Kv.Key), string(evt.Kv.Value))
		}
	}
	return ch
}

func (t *EtcdDriver) WatchPrefix(prefix string) etcdClient.WatchChan {
	return t.etcdClient.Watch(t.ctx, prefix, etcdClient.WithPrefix())
}

func (t *EtcdDriver) SetLease(leaseTime int64) error {
	ctx, cancel := context.WithTimeout(context.TODO(), time.Second*5)
	defer cancel()
	lease := etcdClient.NewLease(t.etcdClient)
	leaseResp, err := lease.Grant(ctx, leaseTime)
	if err != nil {
		return fmt.Errorf("set etcd lease failed")
	}
	t.LeaseID = leaseResp.ID
	t.Lease = lease
	return nil
}

func (t *EtcdDriver) KeepAlive(key string, value string) error {
	resp, err := t.etcdClient.Grant(t.ctx, 5)
	if err != nil {
		fmt.Println("Lease failed,error:", err)
		return err
	}
	_, err = t.etcdClient.Put(t.ctx, key, value, etcdClient.WithLease(resp.ID))
	if err != nil {
		fmt.Println("Lease failed,error:", err)
		return err
	}

	ch, err := t.etcdClient.KeepAlive(t.ctx, resp.ID)
	if err != nil {
		fmt.Println("Lease failed,error:", err)
		return err
	}

	for {
		ka := <-ch
		fmt.Println("ttl:", ka.TTL)
	}
}

func (t *EtcdDriver) registerNode(s *Service, node *Node, opts ...RegisterOption) error {
	if len(s.Nodes) == 0 {
		return fmt.Errorf("require at least one node")
	}

	t.RLock()
	leaseID, ok := t.leases[s.Name+node.Id]
	t.RUnlock()

	if !ok {
		ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
		defer cancel()

		resp, err := t.etcdClient.Get(ctx, nodePath("", s.Name, node.Id), etcdClient.WithSerializable())
		if err != nil {
			return err
		}

		for _, kv := range resp.Kvs {
			if kv.Lease > 0 {
				leaseID = etcdClient.LeaseID(kv.Lease)

				// decode the existing node
				srv := decode(kv.Value)
				if srv == nil || len(srv.Nodes) == 0 {
					continue
				}

				// create hash of service
				h, err := hash.Hash(srv.Nodes[0], nil)
				if err != nil {
					continue
				}

				// save the info
				t.Lock()
				t.leases[s.Name+node.Id] = leaseID
				t.register[s.Name+node.Id] = h
				t.Unlock()

				break
			}
		}
	}

	var leaseNotFound bool

	// renew the lease if it exists
	if leaseID > 0 {
		logrus.Tracef("Renewing existing lease for %s %d", s.Name, leaseID)
		_, err := t.etcdClient.KeepAlive(context.TODO(), leaseID)
		if err != nil {
			return err
		}
		logrus.Tracef("Lease not found for %s %d", s.Name, leaseID)
		// lease not found do register
		leaseNotFound = true
	}

	// create hash of service
	h, err := hash.Hash(node, nil)
	if err != nil {
		return err
	}

	// get existing hash for the service node
	t.Lock()
	v, ok := t.register[s.Name+node.Id]
	t.Unlock()

	// the service is unchanged, skip registering
	if ok && v == h && !leaseNotFound {
		logrus.Tracef("Service %s node %s unchanged skipping registration", s.Name, node.Id)
	}

	service := &Service{
		Name:    s.Name,
		Version: s.Version,
		//Endpoints: s.Endpoints,
		Nodes: []*Node{node},
	}

	var options RegisterOptions
	for _, o := range opts {
		o(&options)
	}

	ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
	defer cancel()

	var lgr *etcdClient.LeaseGrantResponse
	if options.TTL.Seconds() > 0 {
		// get a lease used to expire keys since we have a ttl
		lgr, err = t.etcdClient.Grant(ctx, int64(options.TTL.Seconds()))
		if err != nil {
			return err
		}
	}

	logrus.Tracef("Registering %s id %s with lease %v and ttl %v", service.Name, node.Id, lgr, options.TTL)

	// create an entry for the node
	if lgr != nil {
		_, err = t.etcdClient.Put(ctx, nodePath("", service.Name, node.Id), encode(service), etcdClient.WithLease(lgr.ID))
	} else {
		_, err = t.etcdClient.Put(ctx, nodePath("", service.Name, node.Id), encode(service))
	}

	if err != nil {
		return err
	}

	t.Lock()
	// save our hash of the service
	t.register[s.Name+node.Id] = h
	// save our leaseID of the service
	if lgr != nil {
		t.leases[s.Name+node.Id] = lgr.ID
	}
	t.Unlock()

	return nil
}

func (t *EtcdDriver) Register(s *Service, opts ...RegisterOption) error {
	if len(s.Nodes) == 0 {
		return fmt.Errorf("require at least one node")
	}

	var regerr error

	for _, node := range s.Nodes {
		err := t.registerNode(s, node, opts...)
		if err != nil {
			regerr = err
		}
	}
	return regerr
}

func (t *EtcdDriver) UnRegister(s *Service) error {
	if len(s.Nodes) == 0 {
		return fmt.Errorf("require at least one node")
	}

	for _, node := range s.Nodes {
		t.Lock()
		delete(t.register, s.Name+node.Id)
		delete(t.leases, s.Name+node.Id)
		t.Unlock()

		ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
		defer cancel()

		logrus.Tracef("UnRegistering %s id %s", s.Name, node.Id)
		_, err := t.etcdClient.Delete(ctx, nodePath("", s.Name, node.Id))
		if err != nil {
			return err
		}
	}
	return nil
}

func (t *EtcdDriver) GetService(name string) ([]*Service, error) {
	ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
	defer cancel()
	resp, err := t.etcdClient.Get(ctx, servicePath("", name)+"/", etcdClient.WithPrefix(), etcdClient.WithSerializable())
	if err != nil {
		return nil, fmt.Errorf("get service failed,err: %v", err)
	}
	if len(resp.Kvs) == 0 {
		return nil, fmt.Errorf("service not found")
	}

	serviceMap := map[string]*Service{}

	for _, n := range resp.Kvs {
		if sn := decode(n.Value); sn != nil {
			s, ok := serviceMap[sn.Version]
			if !ok {
				s = &Service{
					Name:    sn.Name,
					Version: sn.Version,
				}
				serviceMap[s.Version] = s
			}

			s.Nodes = append(s.Nodes, sn.Nodes...)
		}
	}

	services := make([]*Service, 0, len(serviceMap))
	for _, service := range serviceMap {
		services = append(services, service)
	}
	return services, nil
}

func (t *EtcdDriver) ListServices() ([]*Service, error) {
	versions := make(map[string]*Service)

	ctx, cancel := context.WithTimeout(t.ctx, time.Duration(t.ReqTimeout)*time.Second)
	defer cancel()

	resp, err := t.etcdClient.Get(ctx, "", etcdClient.WithPrefix(), etcdClient.WithSerializable())
	if err != nil {
		return nil, fmt.Errorf("list services err: %v", err)
	}

	if len(resp.Kvs) == 0 {
		return []*Service{}, nil
	}

	for _, n := range resp.Kvs {
		sn := decode(n.Value)
		if sn == nil {
			continue
		}
		v, ok := versions[sn.Name+sn.Version]
		if !ok {
			versions[sn.Name+sn.Version] = sn
			continue
		}
		v.Nodes = append(v.Nodes, sn.Nodes...)
	}

	services := make([]*Service, 0, len(versions))
	for _, service := range versions {
		services = append(services, service)
	}

	sort.Slice(services, func(i, j int) bool { return services[i].Name < services[j].Name })

	return services, nil
}

func encode(s *Service) string {
	b, _ := json.Marshal(s)
	return string(b)
}

func decode(ds []byte) *Service {
	var s *Service
	_ = json.Unmarshal(ds, &s)
	return s
}

func servicePath(prefix, s string) string {
	return path.Join(prefix, strings.Replace(s, "/", "-", -1))
}

func nodePath(prefix, s, id string) string {
	service := strings.Replace(s, "/", "-", -1)
	node := strings.Replace(id, "/", "-", -1)
	return path.Join(prefix, service, node)
}
